package com.csw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;
import java.util.Scanner;

public class Demo03DianXin {
    private Connection conn;
    private Admin admin;
    @Before
    public void init() throws IOException {
        //创建配置
        Configuration conf = HBaseConfiguration.create();
        //设置hbase  zookeeper连接地址
        conf.set("hbase.zookeeper.quorum","master:2181,node1:2181,node2:2181");
        //创建与hbase的连接
        conn = ConnectionFactory.createConnection(conf);
         admin = conn.getAdmin();
    }
    //将数据加载到HBase里
    @Test
    public void loadDataToHBase()throws Exception{
        //如果表不存在就创建表
        if (!admin.tableExists(TableName.valueOf("dianxin"))){
            admin.createTable(new HTableDescriptor(TableName.valueOf("dianxin")).addFamily(new HColumnDescriptor("info").setMaxVersions(10)));
        }
        //BufferReader读数据
        BufferedReader br = new BufferedReader(new FileReader("E:\\csw\\test\\HBaseAPI\\data\\DIANXIN.csv"));
        //遍历数据
        String line = br.readLine();
        //获取要写入的表
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        while (line!=null){
            String[] split = line.split(",");
            //获取keyrow
            String mdn = split[0];
            String start_time = split[1];
            //经度
            String longitude = split[4];
            //纬度
            String latitude = split[5];
            //以经度和纬度之和作为value
            String pos=longitude+latitude;
            Put put = new Put(mdn.getBytes());
            //使用start_time作为时间戳，利用HBase默认保留时间戳最大的10个版本的数据
            put.addColumn("info".getBytes(),"pos".getBytes(),Long.parseLong(start_time),pos.getBytes());
            dianxin.put(put);
            //获取下一条数据
            line=br.readLine();
        }
    }
    @Test
    public void getLastPostion()throws Exception{
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        String mdn = "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7";
        //设置rowkey
        Get get = new Get(mdn.getBytes());
        // 设置一下需要获取的版本数量，如果不设置，默认获取最新的一条
        get.setMaxVersions(10);
        Result rs = dianxin.get(get);
        //获取RowKer
        byte[] row = rs.getRow();
        String rowkey = Bytes.toString(row);
        System.out.println("rowkey为："+rowkey);
        //使用listCells
        List<Cell> cells = rs.listCells();
        for (Cell cell : cells) {
            //获取列族
            byte[] bytes = CellUtil.cloneFamily(cell);
            String family = Bytes.toString(bytes);
            //获取列名
            byte[] bytes1 = CellUtil.cloneQualifier(cell);
            String qualifier = Bytes.toString(bytes1);
            //获取value
            byte[] bytes2 = CellUtil.cloneValue(cell);
            String value = Bytes.toString(bytes2);

            System.out.println("列族family为："+family+",列qualifier为："+qualifier+",value为："+value);
        }
    }
    @After
    public void close()throws Exception{
        conn.close();
    }
}
